package ${package}.infrastructure.mq;

import ${package}.types.exception.AppException;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Repository;

import java.util.Objects;

/**
 * 使用rocketmq实现
 */
@Repository("rocketmqService")
@Slf4j
public class RocketmqService implements IRocketmqService {

    @Resource
    private RocketMQTemplate rocketmqTemplate;

    /**
     * 检查topic和message是否为空，有一个为空则抛出异常
     *
     * @param topic   主题
     * @param message 消息
     */
    private void checkTopicAndMessageIsBlank(String topic, String message) {
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(message)) {
            throw new AppException("topic:" + topic + " or message:" + message + " is empty");
        }
    }


    @Override
    public void sendMessage(String topic, String message) {
        checkTopicAndMessageIsBlank(topic, message);
        log.info("sendMessage:{} to topic:{}", message, topic);
        rocketmqTemplate.convertAndSend(topic, message);
    }

    @Override
    public void syncSendMessage(String topic, String message) {
        checkTopicAndMessageIsBlank(topic, message);
        log.info("syncSendMessage:{} to topic:{}", message, topic);
        rocketmqTemplate.syncSend(topic, message);
    }

    @Override
    public void asyncSendMessage(String topic, String message) {
        checkTopicAndMessageIsBlank(topic, message);
        log.info("asyncSendMessage:{} to topic:{}", message, topic);
        rocketmqTemplate.asyncSend(topic, message, new SendCallback() {
            /**
             * 发送成功的回调函数
             */
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("asyncSendMessage:{} to topic:{} success:{}", message, topic, sendResult);
            }

            /**
             * 发送失败的回调函数
             */
            @Override
            public void onException(Throwable e) {
                log.error("asyncSendMessage:{} to topic:{} fail", message, topic, e);
            }
        });
    }

    @Override
    public void sendOneWayMessage(String topic, String message) {
        checkTopicAndMessageIsBlank(topic, message);
        log.info("sendOneWay:{} to topic:{}", message, topic);
        rocketmqTemplate.sendOneWay(topic, message);
    }

    @Override
    public void syncSendOrderlyMessage(String topic, String message, String shardingKey) {
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(message) || StringUtils.isBlank(shardingKey)) {
            throw new AppException("topic:" + topic + " or message:" + message + " or shardingKey:" + shardingKey + " is empty");
        }
        log.info("syncSendOrderlyMessage:{} to topic:{} with shardingKey:{}", message, topic, shardingKey);
        rocketmqTemplate.syncSendOrderly(topic, message, shardingKey);
    }

    @Override
    public void sendDelayedMessage(String topic, String message, Integer timeout, Integer delayLevel) {
        checkTopicAndMessageIsBlank(topic, message);
        if (Objects.isNull(timeout) || Objects.isNull(delayLevel) || timeout < 0 || delayLevel < 0) {
            throw new AppException("timeout:" + timeout + " or delayLevel:" + delayLevel + " is null");
        }
        log.info("sendDelayedMessage:{} to topic:{} with timeout:{} and delayLevel:{}", message, topic, timeout, delayLevel);
        rocketmqTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), timeout, delayLevel);
    }
}